/**
 * 
 */
package com.zte.claa.inficombo.csa.app.comm.csif;

import java.nio.charset.Charset;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.mina.core.future.WriteFuture;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zte.claa.inficombo.csa.app.comm.csif.mina.CSIFMinaConnManager;
import com.zte.claa.inficombo.csa.app.comm.csif.mina.msg.CSIFMessage;
import com.zte.claa.inficombo.csa.app.config.CSIFCmdCodeProps;
import com.zte.claa.inficombo.csa.app.config.CSIFCommProps;
import com.zte.claa.inficombo.csa.app.model.app.AppTarget;
import com.zte.claa.inficombo.csa.app.model.redis.SysLog;
import com.zte.claa.inficombo.csa.app.service.redis.CSAMessagePublisher;
import com.zte.claa.inficombo.csa.app.service.redis.PublishDataBuilder;
import com.zte.claa.inficombo.csa.app.util.JsonUtils;

/**
 * @author lyp
 *
 * 针对每一个appeui，实例化一个 CSIFCommService 实例.
 */

public class CSIFCommServiceImpl implements CSIFCommService {
	// 链路检测线程退出通知标记.
	private static final int EXIT_FLAG = -1;	
	// 消息类型定义.
	public static final int MSG_TYPE_UNKNOWN   = -1;
	public static final int MSG_TYPE_RESPONSE  = 0;
	public static final int MSG_TYPE_REPORT    = 1;
	public static final int MSG_TYPE_HEARTBEAT = 2;	
	// 日志打印.
	private Logger loger = LoggerFactory.getLogger(CSIFCommServiceImpl.class);
	// CSIF通讯参数.
	private CSIFCommProps csifCommProps;
	// 命令码/消息码参数.
	private CSIFCmdCodeProps csifCmdCodeProps;
	// 命令码、响应码、消息码
	private Set<String> cmdCodeSet;
	private Set<String> ackCodeSet;
	private Set<String> rptCodeSet;	
	// csif-appeui对象
	private AppTarget appTarget;	
	// 字符集
	private Charset cs = null;	
	// mina底层链路管理器.
	private CSIFMinaConnManager minaConnManager = null;	
	// 链路重连应用层监听.
	private CSIFConnReconnectListener reConnlistener = null;	
	// 上报数据监听器.
	private CSIFReportDataListener rptDataListener = null;
	// cmdseq为键值的map，支持命令并发执行。
	private Map<Integer, CSIFCmdExecContext> execCtxMap = null;
	// redis消息发布器.-->日志上报数据发布；
	private CSAMessagePublisher logPublisher;
	
	// mina链路检查和重建线程池.
	private Thread connChecker = null;
	private LinkedBlockingQueue<Integer> queue = null;
	private boolean isWorkOn = true;
	
	// 最新一次心跳上报时间.
	private long lastHeartBeatTimeMillSec = 0;
	
	/**
	 * 构建器
	 * 
	 * @param appTarget
	 * @param csifCommProps
	 */
	public CSIFCommServiceImpl(AppTarget appTarget, CSIFCommProps csifCommProps){
		this.lastHeartBeatTimeMillSec = System.currentTimeMillis();
		this.appTarget = appTarget;
		this.csifCommProps = csifCommProps;
		this.execCtxMap = new ConcurrentHashMap<Integer, CSIFCmdExecContext>();
		this.queue = new LinkedBlockingQueue<Integer>(6);
		this.isWorkOn = true;
		this.cmdCodeSet = new HashSet<String>();
		this.ackCodeSet = new HashSet<String>();
		this.rptCodeSet = new HashSet<String>();	
	}
	
	/**
	 * 构建器
	 * 
	 * @param appTarget
	 * @param csifCommProps
	 */
	public CSIFCommServiceImpl(AppTarget appTarget){
		this(appTarget, null);
	}
	
	/**
	 * 
	 * @param csifCommProps
	 */
	public void setCSIFCommProps(CSIFCommProps csifCommProps){
		this.csifCommProps = csifCommProps;
	}
	
	/**
	 * 日志发布（主要是csif返回的异步ack响应信息）
	 * @param logPublisher
	 */
	public void setLogPublisher(CSAMessagePublisher logPublisher) {
		this.logPublisher = logPublisher;
	}

	/**
	 * 
	 * @param csifCmdCodeProps
	 */
	public void setCSIFCmdCodeProps(CSIFCmdCodeProps csifCmdCodeProps){
		this.csifCmdCodeProps = csifCmdCodeProps;
		String cmdCodesStr = this.csifCmdCodeProps.getDowncmdcodes();
		String[] cmdCodesArr = cmdCodesStr.split(",");
		this.cmdCodeSet.clear();
		for(String cmdCode : cmdCodesArr){
			this.cmdCodeSet.add(cmdCode);
		}		
		String ackCodeStr = this.csifCmdCodeProps.getAckcmdcodes();
		String[] ackCodeArr = ackCodeStr.split(",");
		this.ackCodeSet.clear();
		for(String ackCode : ackCodeArr){
			this.ackCodeSet.add(ackCode);
		}
		String rptCodeStr = this.csifCmdCodeProps.getRptcmdcodes();
		String[] rptCodeArr = rptCodeStr.split(",");
		this.rptCodeSet.clear();
		for(String rptCode : rptCodeArr){
			this.rptCodeSet.add(rptCode);
		}
	}
	
	/**
	 * 
	 * @return
	 */
	private Charset getCharset(){		
		if(this.cs == null){
			this.cs = Charset.forName(this.csifCommProps.getCharsetname());
		}
		return cs;
	}

	/**
	 * 注册重连监听器.
	 */
	public void setReconnectListener(CSIFConnReconnectListener listener){
		this.reConnlistener = listener;
	}
	
	/**
	 * 设置上报数据监听器  CSIFReportDataListener rptDataListener reConnlistener
	 */
	public void setReportDataListener(CSIFReportDataListener listener) {
		this.rptDataListener = listener;
	}
	
	/**
	 * 获取CSIF目标地址...
	 * @return
	 */
	private String getCSIFIpPort(){
		if(this.appTarget != null){
			return this.appTarget.getAppEUI()+"->"+this.appTarget.getCsifIpAddr()+":"+this.appTarget.getCsifPort();
		}
		return "null";
	}
	
	/* (non-Javadoc)
	 * @see com.zte.claa.infiboss.dataaccess.bossinfcomm.IBossInfCommService#execCommand(java.lang.String)
	 */
	@Override
	public String execCommand(CSIFCommand cmd) throws Exception {		
		CSIFCommand command = cmd;
		try {
			loger.info("[sync]Begin to exec command: " + command+" >>> "+this.getCSIFIpPort());
			CSIFCmdExecContext ctx = new CSIFCmdExecContext(command.getJsonCmdBody());
			this.execCtxMap.put(command.getCmdseq(), ctx);
			CSIFMessage csifCmd = CSIFMessageBuilder.buildMessage(command.getJsonCmdBody(), this.getCharset().name());
			if(this.minaConnManager == null){
				throw new Exception("无法建立到CSIF的连接，请检查项目参数配置: "+this.getCSIFIpPort());
			}
			this.minaConnManager.sendCommand(csifCmd);
			// 等待命令返回（mina.handler上报）或超时.
			this.waitForResult(this.csifCommProps.getCmdtimeoutsecond() * 1000L, ctx);
			// 移除上下文.
			ctx = this.execCtxMap.remove(command.getCmdseq());
			if (ctx == null) {
				throw new Exception("cann't find cmd exec context: " + command);
			}
			String rspMsg = ctx.getRspMsg();
			// 命令响应正常.
			if (rspMsg != null) {
				loger.debug("Command rsp: " + rspMsg);
				return rspMsg;
			}
			String err = "命令执行失败！";
			Throwable t = ctx.getRetErr();
			if (t != null) {
				loger.error("Command exec failed: " + " >>> "+this.getCSIFIpPort(), t);
				err = t.getMessage();
			}
			throw new Exception(err);
		} finally {
			// 命令执行完毕（或超时异常）删除命令执行上下文。
			if(command != null){
				this.execCtxMap.remove(command.getCmdseq());
			}
		}
	}
	
	/**
	 * 执行命令不关注响应.
	 * 
	 * @param jsonCmd
	 * @return 命令响应的string（json）
	 * @throws Exception
	 */
	public WriteFuture execCommandIngoreRsp(CSIFCommand cmd) throws Exception {
		if(cmd == null){
			loger.warn("execCommandIngoreRsp: cmd is null!");
			return null;
		}
		return this.execCommandIngoreRsp(cmd.getJsonCmdBody());
	}
	
	/**
	 * 执行命令不关注响应.
	 * 
	 * @param jsonCmd
	 * @return 命令响应的string（json）
	 * @throws Exception
	 */
	public WriteFuture execCommandIngoreRsp(String jsonCmd) throws Exception {
		loger.info("[async]Begin to exec command: "+jsonCmd+" >>> "+this.getCSIFIpPort());		
		CSIFMessage csifCmd = CSIFMessageBuilder.buildMessage(jsonCmd, this.getCharset().name());
		WriteFuture wf = this.minaConnManager.sendCommand(csifCmd);
		loger.info("Command is sent to CSIF: "+jsonCmd+" >>> "+this.getCSIFIpPort());
		return wf;
	}
	
	/**
	 * 等待执行结果返回。
	 * @param waittime
	 * @throws Exception
	 */
	private void waitForResult(long waittime, CSIFCmdExecContext ctx) throws Exception {
		synchronized(ctx){
			if(ctx.isExecFinished()){
				return;
			}
			ctx.wait(waittime);
		}
	}

	/* (non-Javadoc)
	 * @see com.zte.claa.inficombo.apigwv1.app.framework.service.iotm.IBossInfCommService#reportMessage(java.lang.String)
	 */
	@Override
	public void reportMessage(String msg) {
		if (msg == null) {
			return;
		}
		try {
			int msgtp = this.checkMessageType(msg);
			loger.info("Receive report message from CSIF: tp="+msgtp+", msg=" + msg+", From "+this.getCSIFIpPort());
			switch (msgtp) {
			case MSG_TYPE_RESPONSE:
				this.processCmdResponse(msg);
				break;
			case MSG_TYPE_REPORT:
				this.processUpdataReport(msg);
				break;
			case MSG_TYPE_HEARTBEAT:
				this.processHeartBeatRpt(msg);
				break;
			default:
				loger.warn("Unknow message type: tp=" + msgtp + ", msg=" + msg);
				break;
			}
		} catch (Exception e) {
			loger.error("reportMessage process error: " + msg, e);
		}
	}
	
	/**
	 * 上报消息类型判断
	 * 
	 * @param msg
	 * @return
	 */
	public int checkMessageType(String msg) throws Exception {
		JSONObject json = new JSONObject(msg);
		String cmdCode = JsonUtils.getJsonAttrVal(json, CSIFCmdBuilder.CMD_NAME, null);
		if(cmdCode == null){
			throw new Exception("Invalid msg(no cmd para): "+msg);
		}
		// 由于心跳也属于上报消息的一种，因此这里需要先判断.
		if(cmdCode.equalsIgnoreCase(this.csifCmdCodeProps.getRptheartbeat())){
			return MSG_TYPE_HEARTBEAT;
		}
		if(this.ackCodeSet.contains(cmdCode)){
			return MSG_TYPE_RESPONSE;
		}
		if(this.rptCodeSet.contains(cmdCode)){
			return MSG_TYPE_REPORT;
		}		
		return MSG_TYPE_UNKNOWN;
	}
	
	/**
	 * 处理命令响应消息
	 * 
	 * @param msg
	 */
	private void processCmdResponse(String msg){
		CSIFResponse rsp = null;
		try{
			rsp = CSIFCmdBuilder.parseCSIFRsp(msg);			
			Integer cseq = new Integer(rsp.getCmdseq());
			CSIFCmdExecContext ctx = this.execCtxMap.get(cseq);
			if(ctx == null){
				loger.warn("There is no ctx for msg: "+msg);
				return;
			}
			ctx.setRetMessage(msg);
			synchronized(ctx){
				ctx.notifyAll();
			}
		}catch(Exception e){
			loger.error("processCmdResponse failed: "+msg, e);
		}finally{
			// 将rsp记录到原始日志表，备查...
			if(this.logPublisher != null && rsp != null){
				SysLog log = PublishDataBuilder.buildSysLog("CSIF异步响应上报",
						SysLog.OP_RET_OK, 
						"CSA适配代理通讯框架", 
						rsp.getCmd(), 
						rsp.getJsonRspBody());
				this.logPublisher.publishMessage(JsonUtils.toJson(log));
			}
		}
	}
	
	/**
	 * 处理主动上报消息
	 * 
	 * @param msg
	 */
	private void processUpdataReport(String msg){
		try{
			CSIFReportData rpt = CSIFCmdBuilder.parseCSIFReport(msg);
			// 向上层应用报告上报数据.
			if(this.rptDataListener != null){
				this.rptDataListener.onReport(this.appTarget.getAppEUI(), rpt);
			} else {
				loger.warn("Report data listener is null !!!");
			}
		}catch(Exception e){
			loger.error("processUpdataReport failed: "+msg, e);
		}
	}
	
	/**
	 * 处理心跳消息
	 * 
     *	{
			"cmd": "heartbeat_ack"
		}
	 * 
	 * @param msg
	 */
	private void processHeartBeatRpt(String msg){
		try{
			// 更新最新心跳时间... 
			this.lastHeartBeatTimeMillSec = System.currentTimeMillis();
			String jsonHB = CSIFCmdBuilder.buildHeartBeatAck();
			this.execCommandIngoreRsp(jsonHB);			
		}catch(Exception e){
			loger.error("processHeartBeatRpt failed: "+msg, e);
		}
	}

	/* (non-Javadoc)
	 * @see com.zte.claa.inficombo.apigwv1.app.framework.service.iotm.IotmCommService#reportException(java.lang.Throwable)
	 */
	@Override
	public void reportException(Throwable cause) {
		loger.warn("reportException: "+cause.getMessage());
		// 由于无法识别对应的命令，这里只能匹配当前最新执行的那条命令.
		Integer curCmdSeq = new Integer(CSIFCmdBuilder.getCurrentCmdSeq());
		CSIFCmdExecContext ctx = this.execCtxMap.get(curCmdSeq);
		if(ctx == null){
			loger.warn("There is no cmd exec ctx!");
			return;
		}
		ctx.setRetErr(cause);
		synchronized(ctx){
			ctx.notifyAll();
		}	
	}

	/* (non-Javadoc)
	 * @see com.zte.claa.inficombo.apigwv1.app.framework.service.iotm.IotmCommService#reportConnectStatus(int)
	 */
	@Override
	public void reportConnectStatus(int status) {
		if(status == CSIFCommService.CONN_STATUS_IDLE){
			long curTs = System.currentTimeMillis();
			// 超过3个周期未收到心跳上报...
			if((curTs - this.lastHeartBeatTimeMillSec) >= (this.csifCommProps.getSessionidlesecond()*3*1000L)){
				loger.warn("[!!!]No heartbeat from CSIF over 3 idle periods (210 seconds), reconnect to CSIF: "+this.appTarget);
				this.closeConnectToCSIF();
				try {
					// 建链后会通过重连管理器发起AppEUI的join注册流程... 
					this.connectToCSIF();
				} catch (Exception e) {
					loger.error("connectToCSIF failed: "+this.appTarget, e);
				}
			} else {
				// 通知链路监测线程3秒后启动监测.
				this.queue.offer(new Integer(3));
			}
		}
		if(status == CSIFCommService.CONN_STATUS_CLOSE){
			// 通知链路监测线程20秒后启动监测.
			this.queue.offer(new Integer(20));
		}
	}
	
	/**
	 * 构造心跳消息
	 * 
	 * @return
	 */
	private String buildHeatbeatCmdAck(){
		return CSIFCmdBuilder.buildHeartBeatAck();
	}
	
	/* (non-Javadoc)
	 * @see com.zte.claa.inficombo.apigwv1.app.framework.service.iotm.IBossInfCommService#connectToIotm()
	 */
	@Override
	public void connectToCSIF() throws Exception {
		this.connectToCSIF(false);
	}
	
	/* (non-Javadoc)
	 * @see com.zte.claa.inficombo.apigwv1.app.framework.service.iotm.IotmCommService#hasConnectToIotm()
	 */
	@Override
	public boolean hasConnectToCSIF() {
		if(this.minaConnManager == null){
			return false;
		}
		return true;
	}
	
	/**
	 * 获取appteget
	 * @return
	 */
	public AppTarget getAppTarget() {
		return this.appTarget;
	}

	/**
	 * 建链后60s内需向CSIF进行join认证.
	 */
	@Override
	public String authToCSIF() throws Exception {
		CSIFCommand csJoinCmd = CSIFCmdBuilder.buildCSJoinCmd(this.appTarget);
		return this.execCommand(csJoinCmd);
	}
	
	/**
	 * 向CSIF认证（异步模式）.
	 * 
	 */
	public void authToCSIFByAsynMode() throws Exception {
		CSIFCommand csJoinCmd = CSIFCmdBuilder.buildCSJoinCmd(this.appTarget);
		this.execCommandIngoreRsp(csJoinCmd);
	}
	
	/**
	 * 
	 * @param isAutoReconnect
	 * @throws Exception
	 */
	private synchronized void connectToCSIF(boolean isAutoReconnect) throws Exception {
		// 首次尝试建链就启动链路监测线程.
		if(this.connChecker == null){
			this.isWorkOn = true;
			this.connChecker = new Thread(new Runnable(){
				@Override
				public void run() {
					connectionCheckRun();
				}				
			});
			this.connChecker.setName("CSIFConn-checker-thread");
			this.connChecker.start();
		}
		if(this.minaConnManager != null){
			try{
				this.minaConnManager.releaseConnection();
				this.minaConnManager = null;
			}catch(Exception e){
				loger.warn(e.getMessage());
			}
		}
		this.minaConnManager = new CSIFMinaConnManager(this.csifCommProps, this.appTarget, this);
		try{
			this.minaConnManager.createConnect();
		}catch(Exception ex){
			this.minaConnManager = null;
			this.queue.offer(new Integer(20));
			throw ex;
		}
		loger.info("connect to CSIF ok: "+this.getCSIFIpPort());
		if(this.reConnlistener != null && isAutoReconnect){
			this.reConnlistener.onReconnect(this.appTarget.getAppEUI());
		}
		// 重建连接后，以当前时间更新最新心跳时间.
		this.lastHeartBeatTimeMillSec = System.currentTimeMillis();
	}

	/* (non-Javadoc)
	 * @see com.zte.claa.inficombo.apigwv1.app.framework.service.iotm.IBossInfCommService#closeConnectToIotm()
	 */
	@Override
	public synchronized void closeConnectToCSIF() {
		// 停止链路监测线程.
		this.isWorkOn = false;
		this.queue.clear();
		this.queue.offer(new Integer(EXIT_FLAG));
		this.connChecker = null;
		// 关闭链路》
		if(this.minaConnManager == null){
			return;
		}
		try{
			this.minaConnManager.releaseConnection();			
			loger.warn("close connection to CSIF ok: "+this.getCSIFIpPort());
		}catch(Exception e){
			loger.warn(e.getMessage());
		}finally{
			this.minaConnManager = null;
		}
	}
	
	/**
	 * 链路检查线程的线程体方法
	 * 
	 */
	private void connectionCheckRun(){
		while (isWorkOn) {
			loger.info("connection check thread is running: "+this.getCSIFIpPort());
			try {
				Integer delaySec = this.queue.poll(60, TimeUnit.SECONDS);
				if(delaySec == null){
					continue;
				}
				if(delaySec.intValue() == EXIT_FLAG){
					break;
				} else {
					this.connectionCheck(delaySec.intValue());
				}
			} catch (Exception e) {
			}
		}
		loger.warn("connection checker exit: "+this.getCSIFIpPort());
	}
	
	/**
	 * 链路监测。
	 */
	private void connectionCheck(int delaySeconds){
		try{
			Thread.sleep(delaySeconds*1000L);
		}catch(Exception ignore){}
		if(this.isWorkOn == false){
			loger.warn("isWorkOn flag is false, begin exit!");
			return;
		}
		String hbCmd = this.buildHeatbeatCmdAck();
		try{
			// 通过发送心跳命令检查链路.
			WriteFuture wf = this.execCommandIngoreRsp(hbCmd);			
			wf.awaitUninterruptibly(this.csifCommProps.getCmdsendtimeoutsecond() * 1000L);
			if(!wf.isWritten()){
				String err = "消息下发超时！";
				Throwable t = wf.getException();
				if(t != null){
					err = t.getMessage();
				}
				throw new Exception("消息下发失败："+err+", cmd=["+hbCmd.toString()+"] >>> "+this.getCSIFIpPort());
			}
		}catch(Exception e){
			if(this.isWorkOn == false){
				loger.warn("isWorkOn flag is false, begin exit!");
				return;
			}
			loger.error("Connection is break and try to reconnect: "+e.getMessage()+" >>> "+this.getCSIFIpPort());
			try {
				this.connectToCSIF(true);	
			} catch (Exception ex) {
				if(this.queue.size() <= 0){
					this.queue.offer(new Integer(60));
				}
			}
		}
	}
	
}
